package com.hsh.db;

import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.HashMap;
import java.util.Map;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.db.DBInputFormat;

public class MyRecordReader extends RecordReader<LongWritable, MySqlDBWriteAble> {
	
	private DBInputFormat.DBInputSplit split;
	
	private ResultSet resultSet;
	
	private String[] columns;
	
	private String queryCondition;
	
	private String tableName;
	
	private LongWritable key;
	
	private MySqlDBWriteAble value;
	
	private Connection conn;
	
	
	Log LOG = LogFactory.getLog(MyRecordReader.class);
	
	public MyRecordReader(InputSplit split) {
		super();
		this.split = (DBInputFormat.DBInputSplit)split;
	}
	
	@Override
	public void initialize(InputSplit split, TaskAttemptContext context)
			throws IOException, InterruptedException {
		try {
			Configuration conf = context.getConfiguration();
			this.tableName = conf.get("tableName");
			this.columns = conf.get("columns").split(",");
			key = new LongWritable(0);
			
			Class.forName("com.mysql.jdbc.Driver");
			conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/test", "root", "root");
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	@Override
	public boolean nextKeyValue() throws IOException, InterruptedException {
		if(resultSet == null){
			resultSet = executeQuery();
		}
		try {
			if(!resultSet.next())
				return false;
		} catch (SQLException e) {
			e.printStackTrace();
		}
		key = new LongWritable(key.get()+1);
		value = new MySqlDBWriteAble();
		value.setInputColNames(columns);
		try {
			value.readFields(resultSet);
		} catch (SQLException e) {
			e.printStackTrace();
		}
		return true;
	}

	private ResultSet executeQuery() {
		String sql = buildSql();
		try {
			Statement stmt = conn.createStatement();
			return stmt.executeQuery(sql);
		} catch (Exception e) {
			e.printStackTrace();
		}
		return null;
	}
	
	private String buildSql(){
		StringBuffer sql = new StringBuffer(20);
		sql.append("SELECT ");
		if(columns != null && columns.length > 0){
			String columnStr = "";
			for(String clo : columns){
				columnStr += (clo + ",");
			}
			sql.append(columnStr.substring(0, columnStr.length()-1));
		}else{
			sql.append(" * ");
		}
		sql.append(" FROM ").append(tableName);
		if(queryCondition != null)
			sql.append(queryCondition).append(";");
		
		try {
			sql.append(" LIMIT ").append(split.getStart()).append(",").append(split.getLength()).append(";");
		} catch (IOException e) {
			e.printStackTrace();
		}
		LOG.info(sql.toString());
		return sql.toString();
	}
	
	private Map<String,String> createValue(){
		Map<String, String> rtn = new HashMap<String, String>(4);
		rtn.put("col1", System.currentTimeMillis()+"");
		return rtn;
	}

	@Override
	public LongWritable getCurrentKey() throws IOException,
			InterruptedException {
		return key;
	}

	@Override
	public MySqlDBWriteAble getCurrentValue() throws IOException, InterruptedException {
		return value;
	}

	@Override
	public float getProgress() throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		return 0;
	}

	@Override
	public void close() throws IOException {
		try{
		if(resultSet != null)
			resultSet.close();
		
		if(conn != null)
			conn.close();
		}catch (SQLException exp){
			exp.printStackTrace();
		}
	}

}
